package com.atguigu.flink.chapter02_DataStreamAPI.process;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Smexy on 2022/10/22
 *
 *      按照传感器的id,求同一种传感器的水位和.
 *
 *              使用Flink提供的状态来解决。
 *
 *
 *
 */
public class Demo14_ProcessByKey
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
            .socketTextStream("hadoop103", 8888)
            .map(new MapFunction<String, WaterSensor>()
           {
               @Override
               public WaterSensor map(String value) throws Exception {
                   String[] data = value.split(",");
                   return new WaterSensor(
                       data[0],
                       Long.valueOf(data[1]),
                       Integer.valueOf(data[2])
                   );
               }
           })
            .keyBy(WaterSensor::getId)
            .process(new KeyedProcessFunction<String, WaterSensor, String>()
            {
                Map<String,Integer> sums = new HashMap<>();
                //KeyedProcessFunction<K, I, O>
                @Override
                public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                    //获取当前的key
                    String currentKey = ctx.getCurrentKey();
                    //从map中获取当前key已经累加的值
                    // 如果取不到，说明当前这条数据，是这个key的第一个，设置为0
                    Integer valueBefore = sums.getOrDefault(currentKey, 0);

                    //累加
                    valueBefore += value.getVc();

                    //放入map，供后续的数据继续累加
                    sums.put(currentKey,valueBefore);

                    out.collect(currentKey + ":" + valueBefore);

                }
            }).setParallelism(2)
            .print();


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
